/**
 * Copyright (c) 2013-2020 Nikita Koksharov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.pubsub.LockPubSub;

/**
 * Lock will be removed automatically if client disconnects.
 *
 * @author Nikita Koksharov
 *
 */
public class RedissonReadLock extends RedissonLock implements RLock {

    public RedissonReadLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
    }

    @Override
    String getChannelName() {
        return prefixName("redisson_rwlock", getName());
    }
    
    String getWriteLockName(long threadId) {
        return super.getLockName(threadId) + ":write";
    }

    String getReadWriteTimeoutNamePrefix(long threadId) {
        return suffixName(getName(), getLockName(threadId)) + ":rwlock_timeout"; 
    }

    @Override
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        /**
         * 加读锁的lua脚本
         * keys[1] anyRwLock
         * keys[2] {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout
         * ARGV[1] 30000
         * ARGV[2] UUID_01:ThreadId_01
         * ARGV[3] UUID_01:ThreadId_01:write
         *
         * 第一种情况读读:
         * 第一个线程ThreadId_01
         * hget anyRwLock mode 返回 false
         * hset anyRwLock mode read
         * hset anyRwLock UUID_01:ThreadId_01 1
         * set {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         * pexpire UUID_01:ThreadId_01:1 30000
         * pexpire anyRwLock 30000
         * 此时anyRwLock数据结构为
         * {
         *     "mode":"read",
         *     "UUID_01:ThreadId_01":1
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         * 两者过期时间都为30s
         *
         * 第二个线程ThreadId_02
         * hget anyRwLock mode 返回 read
         * hincrby anyRwLock UUID_02:ThreadId_02 1 返回 ind=1
         * set {anyRwLock}:UUID_02:ThreadId_02:rwlock_timeout:1 1
         * pexpire {anyRwLock}:UUID_02:ThreadId_02:rwlock_timeout:1 30000
         * 假设 anyRwLock 过期时间为20s
         * pexpire anyRwLock 30000
         * 返回nil
         *
         * 此时数据结构为
         * {
         *     "mode":"read",
         *     "UUID_01:ThreadId_01":1,
         *     "UUID_02:ThreadId_02":1
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         * {anyRwLock}:UUID_02:ThreadId_02:rwlock_timeout:1 1
         *
         *
         * 第二种情况写读不同客户端(阻塞):
         * 此时anyRwLock数据结构为
         * {
         *     "mode":"write",
         *     "UUID_01:ThreadId_01:write":1
         * }
         * hget anyRwLock mode 返回为write
         * hexists UUID_02:ThreadId_02:write 返回为0
         * 返回 pttl anyRwLock
         *
         * 第三种情况读写相同线程
         * 此时anyRwLock数据结构为
         * {
         *     "mode":"write",
         *     "UUID_01:ThreadId_01:write":1
         * }
         * hget anyRwLock mode 返回为write
         * hexists UUID_01:ThreadId_01:write 返回为1
         * hincrby anyRwLock UUID_01:ThreadId_01 1 返回 ind=1
         * set {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         * pexpire {anyRwLock}:UUID_02:ThreadId_02:rwlock_timeout:ind 30000
         * anyLock和上面读锁key过期时间都为30s
         * 此时数据结构为
         * {
         *     "mode":"write",
         *     "UUID_01:ThreadId_01:write":1,
         *     "UUID_01:ThreadId_01":1
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         */
        return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                                "local mode = redis.call('hget', KEYS[1], 'mode'); " +
                                "if (mode == false) then " +
                                  "redis.call('hset', KEYS[1], 'mode', 'read'); " +
                                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                                  "redis.call('set', KEYS[2] .. ':1', 1); " +
                                  "redis.call('pexpire', KEYS[2] .. ':1', ARGV[1]); " +
                                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                  "return nil; " +
                                "end; " +
                                "if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " +
                                        // 此时有人加了read或者是自己加了write
                                  "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " + 
                                  "local key = KEYS[2] .. ':' .. ind;" +
                                  "redis.call('set', key, 1); " +
                                  "redis.call('pexpire', key, ARGV[1]); " +
                                  "local remainTime = redis.call('pttl', KEYS[1]); " +
                                  "redis.call('pexpire', KEYS[1], math.max(remainTime, ARGV[1])); " +
                                  "return nil; " +
                                "end;" +
                                "return redis.call('pttl', KEYS[1]);",
                        Arrays.<Object>asList(getName(), getReadWriteTimeoutNamePrefix(threadId)), 
                        internalLockLeaseTime, getLockName(threadId), getWriteLockName(threadId));
    }

    @Override
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
        String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);

        /**
         * 读锁释放参数:
         * KEYS[1] anyRwLock
         * KEYS[2] {redisson_rwlock}:UUID_01:ThreadId_01
         * KEYS[3] {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout
         * KEYS[4] {anyRwLock}
         * ARGV[1] 0L
         * ARGV[2] UUID_01:ThreadId_01
         *
         * 第一种情况 一个线程获取了读锁此时数据结构
         * {
         *     "mode":"read",
         *     "UUID_01:ThreadId_01":1
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         *
         * hget anyRwLock mode 返回read
         * hexists anyRwLock UUID_01:ThreadId_01 返回 lockExists=1
         * hincrby anyRwLock UUID_01:ThreadId_01 -1 返回 counter=0
         * hdel anyRwLock UUID_01:ThreadId_01
         * del {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1
         * hlen anyRwLock 返回为1只有mode
         * del anyRwLock
         * 返回 1 结束
         *
         * 第二种情况 多次获取读锁此时数据结构为
         * {
         *     "mode":"read",
         *     "UUID_01:ThreadId_01":2
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:2 1
         *
         * hget anyRwLock mode 返回read
         * hexists anyRwLock UUID_01:ThreadId_01 返回 lockExists=2
         * hincrby anyRwLock UUID_01:ThreadId_01 -1 返回 counter=1
         * del {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:2
         * 此时数据结构为
         * {
         *     "mode":"read",
         *     "UUID_01:ThreadId_01":1
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         *
         * hlen anyRwLock 返回结果2
         * hkeys anyRwLock 遍历获取所有的key  key= UUID_01:ThreadId_01
         * hget anyRwLock UUID_01:ThreadId_01 返回为1
         * counter 从1遍历到0
         * pttl {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 返回remainTime假设为20s 即第二个锁还有20s过期
         * maxRemainTime = 20s(如果过期默认为-3)
         * pexpire anyRwLock 20000
         * 返回0结束代表锁还未释放完
         *
         * 第三种情况 写锁读锁此时anyRwLock数据结构为
         * {
         *     "mode":"write",
         *     "UUID_01:ThreadId_01:write":1,
         *     "UUID_01:ThreadId_01":1
         * }
         * {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 1
         *
         * 步骤同上看不同地方
         * 被删除后:
         * {
         *     "mode":"write",
         *     "UUID_01:ThreadId_01:write":1,
         * }
         *
         * pttl {anyRwLock}:UUID_01:ThreadId_01:rwlock_timeout:1 此时返回为-2
         * mode =write 返回0 认为还有线程获取锁
         */
        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +
                "if (mode == false) then " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end; " +
                "local lockExists = redis.call('hexists', KEYS[1], ARGV[2]); " +
                "if (lockExists == 0) then " +
                    "return nil;" +
                "end; " +
                    
                "local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); " + 
                "if (counter == 0) then " +
                    "redis.call('hdel', KEYS[1], ARGV[2]); " + 
                "end;" +
                "redis.call('del', KEYS[3] .. ':' .. (counter+1)); " +
                
                "if (redis.call('hlen', KEYS[1]) > 1) then " +
                    "local maxRemainTime = -3; " + 
                    "local keys = redis.call('hkeys', KEYS[1]); " + 
                    "for n, key in ipairs(keys) do " + 
                        "counter = tonumber(redis.call('hget', KEYS[1], key)); " + 
                        "if type(counter) == 'number' then " + 
                            "for i=counter, 1, -1 do " + 
                                "local remainTime = redis.call('pttl', KEYS[4] .. ':' .. key .. ':rwlock_timeout:' .. i); " + 
                                "maxRemainTime = math.max(remainTime, maxRemainTime);" + 
                            "end; " + 
                        "end; " + 
                    "end; " +
                            
                    "if maxRemainTime > 0 then " +
                        "redis.call('pexpire', KEYS[1], maxRemainTime); " +
                        "return 0; " +
                    "end;" + 
                        
                    "if mode == 'write' then " + 
                        "return 0;" + 
                    "end; " +
                "end; " +
                    
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; ",
                Arrays.<Object>asList(getName(), getChannelName(), timeoutPrefix, keyPrefix), 
                LockPubSub.UNLOCK_MESSAGE, getLockName(threadId));
    }

    protected String getKeyPrefix(long threadId, String timeoutPrefix) {
        return timeoutPrefix.split(":" + getLockName(threadId))[0];
    }
    
    @Override
    protected RFuture<Boolean> renewExpirationAsync(long threadId) {
        String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
        String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);
        
        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "local counter = redis.call('hget', KEYS[1], ARGV[2]); " +
                "if (counter ~= false) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                    
                    "if (redis.call('hlen', KEYS[1]) > 1) then " +
                        "local keys = redis.call('hkeys', KEYS[1]); " + 
                        "for n, key in ipairs(keys) do " + 
                            "counter = tonumber(redis.call('hget', KEYS[1], key)); " + 
                            "if type(counter) == 'number' then " + 
                                "for i=counter, 1, -1 do " + 
                                    "redis.call('pexpire', KEYS[2] .. ':' .. key .. ':rwlock_timeout:' .. i, ARGV[1]); " + 
                                "end; " + 
                            "end; " + 
                        "end; " +
                    "end; " +
                    
                    "return 1; " +
                "end; " +
                "return 0;",
            Arrays.<Object>asList(getName(), keyPrefix), 
            internalLockLeaseTime, getLockName(threadId));
    }
    
    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }

    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hget', KEYS[1], 'mode') == 'read') then " +
                    "redis.call('del', KEYS[1]); " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end; " +
                "return 0; ",
                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE);
    }

    @Override
    public boolean isLocked() {
        RFuture<String> future = commandExecutor.writeAsync(getName(), StringCodec.INSTANCE, RedisCommands.HGET, getName(), "mode");
        String res = get(future);
        return "read".equals(res);
    }

}
